package com.easydev.mq.rabbit.pool;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeoutException;

import org.apache.commons.pool2.BasePooledObjectFactory;
import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.impl.DefaultPooledObject;

import com.easydev.mq.exception.MessageException;
import com.easydev.mq.rabbit.RabbitMQConfig;
import com.rabbitmq.client.Channel;

public class RabbitChannelPoolFactory extends BasePooledObjectFactory<Channel> {

    private List<RabbitConnection> connections;

    private static RabbitChannelPoolFactory instance;

    private int PRE_TOTLE;

    private int CON_TOTLE;

    private String virtualHost;

    private RabbitChannelPoolFactory(String virtualHost) throws MessageException {
        CON_TOTLE = RabbitMQConfig.CONNECTION_NUM;
        PRE_TOTLE = RabbitMQConfig.CHANNEL_PER_CONNECTION;
        this.virtualHost = virtualHost;
        connections = new ArrayList<RabbitConnection>();
        // connections.add(new RabbitConnection(virtualHost));
    }

    /**
     * Description: 创建Channel，如果connection中channel数量超出预设阈值，创建connection<br>
     * 
     * @author lx<br>
     * @return <br>
     * @throws MessageException
     */
    private synchronized Channel createChannel() throws MessageException {
        Channel channel = null;
        boolean flag = false;
        for (RabbitConnection connection : connections) {
            if (connection.size() < PRE_TOTLE) {
                channel = connection.createChannel();
                flag = true;
                break;
            }
        }
        if (!flag && connections.size() < CON_TOTLE) {
            RabbitConnection newConnection = new RabbitConnection(virtualHost);
            channel = newConnection.createChannel();
            connections.add(newConnection);
            flag = true;
        }
        if (!flag) {
            throw new MessageException(
                    "CREATE CHANNEL FAILED，PLEASE CHECK CONFIG，CHANNEL NUM CAN NOT BIGGER THAN PRE_TOTLE*CON_TOTLE");
        }
        return channel;
    }

    /**
     * Description: 删除对应CONNECTION中的CHANNEL 并关闭CHENNEL<br>
     * 
     * @author lx<br>
     * @param channel
     * @throws IOException
     * @throws TimeoutException <br>
     */
    private synchronized void deleteChannel(Channel channel) throws IOException, TimeoutException {
        List<RabbitConnection> delConnection = new ArrayList<RabbitConnection>();
        for (int i = 0; i < connections.size(); i++) {
            RabbitConnection connection = connections.get(i);
            boolean deleteFlag = false;
            for (Channel channel_con : connection.getSubChannels()) {
                if (channel.equals(channel_con)) {
                    deleteFlag = true;
                }
            }
            if (deleteFlag) {
                if (channel.isOpen()) {
                    channel.close();
                }
                connection.remove(channel);
            }
            if (connection.size() == 0) {
                delConnection.add(connection);
            }
        }
        for (RabbitConnection connection : delConnection) {
            if (connection.valid()) {
                connection.colse();
            }
            connections.remove(connection);
        }
        delConnection = new ArrayList<RabbitConnection>();
    }

    /**
     * Description: RabbitChannelPoolFactory单例方法<br>
     * 
     * @return instance
     * @throws MessageException
     */
    public static synchronized RabbitChannelPoolFactory getInstance(String virtualHost) throws MessageException {
        if (instance == null) {
            instance = new RabbitChannelPoolFactory(virtualHost);
        }
        return instance;
    }

    @Override
    public Channel create() throws Exception {
        return createChannel();
    }

    @Override
    public PooledObject<Channel> wrap(Channel obj) {
        return new DefaultPooledObject<Channel>(obj);
    }

    @Override
    public void destroyObject(PooledObject<Channel> p) throws Exception {
        Channel channel = p.getObject();
        deleteChannel(channel);
        super.destroyObject(p);
    }

    @Override
    public boolean validateObject(PooledObject<Channel> p) {
        Channel channel = p.getObject();
        if (channel.isOpen()) {
            return super.validateObject(p);
        } else {
            return false;
        }
    }

}
